package com.zjh.storm.bolts;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by zhoujianhua on 2017/9/6.
 */
public class WordCounterBolt extends BaseBasicBolt{
    Integer id;
    String name;
    Map<String, Integer> counters;

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        counters = new HashMap<String,Integer>();
        this.name = context.getThisComponentId();
        this.id = context.getThisTaskId();
    }

    @Override
    public void cleanup() {
        System.out.println("-- Word counter [" + name + "-" + id + "] --");
        for (Map.Entry<String, Integer> stringIntegerEntry : counters.entrySet()) {
            System.out.println(stringIntegerEntry.getKey() + ": " + stringIntegerEntry.getValue());
        }
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        String str = input.getString(0);
        if (!counters.containsKey(str)) {
            counters.put(str, 1);
        } else {
            int c = counters.get(str) + 1;
            counters.put(str, c);
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
